package com.akka.demo.actor;

import akka.actor.*;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.japi.pf.DeciderBuilder;
import com.akka.demo.di.SpringExtension;
import com.akka.demo.model.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import scala.concurrent.duration.Duration;
import scala.concurrent.duration.FiniteDuration;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static akka.actor.SupervisorStrategy.*;

/**
 * Created by Administrator on 2018/2/26.
 */
@Component
@Scope("prototype")
public class IotSupervisor extends AbstractActor {

    private LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);

    @Autowired
    private SpringExtension springExtension;

    @Override
    public void preStart() {
        log.info("IoT Application started");
    }

    @Override
    public void postStop() {
        log.info("IoT Application stopped");
    }


    //maxNrOfRetries设置-1同时withinTimeRange设置无限(Duration.Inf())则无限重启
    //maxNrOfRetries设置非负数，同时withinTimeRange设置无限则一旦重启次数超过maxNrOfRetries则子actor会stop
    private static SupervisorStrategy strategy =    //子actor 1分钟内如果重启超过10次则会Stop
            new OneForOneStrategy(10, Duration.create(1, TimeUnit.MINUTES), DeciderBuilder.
                    match(ArithmeticException.class, e -> resume()).
                    match(NullPointerException.class, e -> restart()).
                    match(IllegalArgumentException.class, e -> stop()).
                    matchAny(o -> escalate()).build());

    @Override
    public SupervisorStrategy supervisorStrategy() {
        return strategy;
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .matchEquals("start", x -> {
                    ActorRef deviceActor = context().actorOf(springExtension.props("device", "group", "device"), "child-device-actor");
                    deviceActor.tell(new RecordTemperature(1L, 24.02), getSelf());
                    deviceActor.tell(new ReadTemperature(2L), getSelf());
                })
                .match(TemperatureRecorded.class, temperatureRecorded -> {
                    log.info("Temperature Recorded, requestId is {}", temperatureRecorded.getRequestId());
                })
                .match(RespondTemperature.class, respondTemperature -> {
                    log.info("Temperature is {},requestId is {}", respondTemperature.getValue(), respondTemperature.getRequestId());
                })
                .matchEquals("initGroup", x -> {
                    ActorRef deviceManagerActor = context().actorOf(springExtension.props("deviceManager"), "deviceManager-actor");
                    deviceManagerActor.tell(new DeviceManager.RequestTrackDevice("group1", "device1"), getSelf());
                })
                .match(DeviceManager.DeviceRegistered.class, deviceRegistered -> {
                    log.info("Device registered!!! --- from IotSupervisor");
                })
                .match(DeviceGroup.RequestAllTemperatures.class, x -> {
                    Map<ActorRef, String> actorToDeviceId = new HashMap<>();
                    ActorRef deviceActor1 = context().actorOf(springExtension.props("device", "group1", "device1"), "device-actor-1");
                    ActorRef deviceActor2 = context().actorOf(springExtension.props("device", "group2", "device2"), "device-actor-2");
                    actorToDeviceId.put(deviceActor1, "device1");
                    actorToDeviceId.put(deviceActor2, "device2");
                    context().actorOf(springExtension.props("deviceGroupQuery",
                            actorToDeviceId, 1L, self(), FiniteDuration.create(3L, TimeUnit.SECONDS)), "deviceGroupQuery-actor");
                })
                .match(DeviceGroup.RespondAllTemperatures.class, respondAllTemperatures -> {
                    long requestId = respondAllTemperatures.getRequestId();
                    Map<String, TemperatureReading> map = respondAllTemperatures.getTemperatures();
                    for (Map.Entry<String, TemperatureReading> entry : map.entrySet()) {
                        String deviceId = entry.getKey();
                        TemperatureReading reading = entry.getValue();
                        if (reading instanceof DeviceGroup.Temperature) {
                            double value = ((DeviceGroup.Temperature) reading).getValue();
                            log.info("Device {} response temperature is {} with requestId {}", deviceId, value, requestId);
                        } else if (reading instanceof DeviceGroup.TemperatureNotAvailable) {
                            log.info("Device {} response temperature is not available with requestId {}", deviceId, requestId);
                        } else if (reading instanceof DeviceGroup.DeviceNotAvailable) {
                            log.info("Device {} is not available with requestId {}", deviceId, requestId);
                        } else if (reading instanceof DeviceGroup.DeviceTimedOut) {
                            log.info("Device {} timed out with requestId {}", deviceId, requestId);
                        }
                    }
                })
                .build();
    }
}
